package com.googlecode.objectify.hbase;

import com.google.appengine.api.datastore.AsyncDatastoreService;
import com.google.appengine.api.datastore.DatastoreAttributes;
import com.google.appengine.api.datastore.DatastoreService.KeyRangeState;
import com.google.appengine.api.datastore.DatastoreServiceConfig;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.EntityNotFoundException;
import com.google.appengine.api.datastore.Index;
import com.google.appengine.api.datastore.Index.IndexState;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyFactory;
import com.google.appengine.api.datastore.KeyRange;
import com.google.appengine.api.datastore.PreparedQuery;
import com.google.appengine.api.datastore.Query;
import com.google.appengine.api.datastore.Transaction;
import com.google.appengine.api.datastore.TransactionOptions;
import com.google.common.util.concurrent.MoreExecutors;
import com.googlecode.objectify.generic.IdRangeAllocator;
import fi.aalto.hacid.HAcidClient;
import fi.aalto.hacid.HAcidDelete;
import fi.aalto.hacid.HAcidGet;
import fi.aalto.hacid.HAcidPut;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.TableNotFoundException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

/**
 *
 * @author Secondary User
 */
public class HBaseAsyncDatastoreService implements AsyncDatastoreService, IdRangeAllocator
{

	private final DatastoreServiceConfig config;
	private HTablePool pool;
	private static int poolSize = 20;
	private static int poolThreads = 5;
	private ExecutorService exec = Executors.newFixedThreadPool(poolThreads);
	HAcidClient client;
	static final byte[] FAMILY = Bytes.toBytes("f");
	private HBaseTransaction currentTxn;
	private List<HBaseTransaction> transactions = new ArrayList<HBaseTransaction>();
	private HBaseIndexer indexer;

	//TODO: Implement this class (and some results such as Transaction, Query, etc.)
	public HBaseAsyncDatastoreService(DatastoreServiceConfig config) throws Exception
	{
		this.config = config;
		pool = new HTablePool(config.getConfig(), poolSize, new HAcidHTableFactory());
		client = new HAcidClient(config.getConfig());
		indexer = new HBaseIndexer(this);
		MoreExecutors.addDelayedShutdownHook(exec, 48, TimeUnit.HOURS);
	}

	private HTableInterface main()
	{
		return getTable(config.getApplicationName() + "M");
	}

	private HTableInterface getTable(String tableName)
	{
		try
		{
			return pool.getTable(tableName);
		}
		catch (Exception e)
		{
			if (e instanceof TableNotFoundException || (e.getCause() != null && e.getCause() instanceof TableNotFoundException))
			{
				HTableDescriptor ht = new HTableDescriptor(tableName);
				ht.addFamily(new HColumnDescriptor(FAMILY));
				HBaseAdmin hba;
				try
				{
					hba = new HBaseAdmin(config.getConfig());
					hba.createTable(ht);
					return pool.getTable(tableName);
				}
				catch (MasterNotRunningException ex)
				{
					throw new RuntimeException(ex);
				}
				catch (ZooKeeperConnectionException ex)
				{
					throw new RuntimeException(ex);
				}
				catch (IOException ex)
				{
					throw new RuntimeException(ex);
				}
			}
			else
			{
				throw new RuntimeException(e);
			}
		}
	}

	private HTableInterface ids()
	{
		return getTable(config.getApplicationName() + "N"); //N for numbers
	}

	HTableInterface index()
	{
		return getTable(config.getApplicationName() + "I");
	}

	@Override
	public Future<Transaction> beginTransaction(TransactionOptions options)
	{
		return beginTransaction(options, false);
	}

	private Future<Transaction> beginTransaction(TransactionOptions options, final boolean autocommit)
	{
		return exec.submit(new Callable<Transaction>()
		{
			@Override
			public Transaction call() throws Exception
			{
				return new HBaseTransaction(config.getApplicationName(), exec, client, autocommit);
			}
		});
	}

	@Override
	public Future<KeyRange> allocateIds(String kind, long num)
	{
		return allocateIds(null, kind, num);
	}

	@Override
	public Future<KeyRange> allocateIds(final Key parent, final String kind, final long num)
	{
		//Transactions not needed because HBase already does row-level transactions.
		return exec.submit(new Callable<KeyRange>()
		{
			@Override
			public KeyRange call() throws Exception
			{
				String key;
				if (parent != null)
				{
					key = parent.toKeyString();
				}
				else
				{
					key = "";
				}
				key += Key.SPLIT_CHAR + kind;
				Increment inc = new Increment(Bytes.toBytes(key));
				byte[] idBytes = Bytes.toBytes("id");
				inc.addColumn(FAMILY, idBytes, num);
				Result r = ids().increment(inc);
				byte[] lastId = r.getValue(FAMILY, idBytes);
				long lastIdL;
				if (lastId == null)
				{
					lastIdL = -1;
				}
				else
				{
					lastIdL = Bytes.toLong(lastId);
				}
				List<Key> keyList = new ArrayList<Key>();
				for (long id = lastIdL - num; id <= lastIdL; id++)
				{
					keyList.add(KeyFactory.createKey(parent, kind, id));
				}
				return new KeyRange(keyList);
			}
		});
	}

	@Override
	public Future<Void> delete(Key... keys)
	{
		return delete(Arrays.asList(keys));
	}

	@Override
	public Future<Void> delete(Iterable<Key> keys)
	{
		return delete(null, keys);
	}

	@Override
	public Future<Void> delete(Transaction txn, Key... keys)
	{
		return delete(txn, Arrays.asList(keys));
	}

	@Override
	public Future<Void> delete(final Transaction txn, final Iterable<Key> keys)
	{
		//TODO: Update currentTxn if passed in here???
		return exec.submit(new Callable<Void>()
		{
			@Override
			public Void call() throws Exception
			{
				for (Key key : keys)
				{
					if (key == null)
					{
						continue;
					}
					//TODO: Chunking?
					HBaseTransaction thisTxn = checkTransaction(txn);
					Entity e = get(thisTxn, key).get();
					//TODO: thisTxn.delete(new HAcidDelete(main(), Bytes.toBytes(key.toKeyString())));
					deleteFromAllIndexes(e);
					if (txn != thisTxn)
					{
						thisTxn.commit();
					}
				}
				return null;
			}
		});
	}

	private void deleteFromAllIndexes(final Entity e)
	{
		exec.submit(new Runnable()
		{
			@Override
			public void run()
			{
				indexer.deleteFromIndexes(e);
			}
		});
	}

	private HBaseTransaction checkTransaction(Transaction txn) throws InterruptedException, ExecutionException
	{
		HBaseTransaction thisTxn = null;
		if (txn != null && txn instanceof HBaseTransaction && txn.isActive())
		{
			thisTxn = (HBaseTransaction) txn;
		}
		if (thisTxn == null || !thisTxn.isActive())
		{
			thisTxn = (HBaseTransaction) getCurrentTransaction();
		}
		if (thisTxn == null || !thisTxn.isActive())
		{
			thisTxn = (HBaseTransaction) beginTransaction(null, true).get();
			//TODO: Update currentTxn here??
			//TODO: Update transactions here?
		}
		return thisTxn;
	}

	@Override
	public Future<Entity> get(Key key)
	{
		return get(null, key);
	}

	@Override
	public Future<Map<Key, Entity>> get(Iterable<Key> keys)
	{
		return get(null, keys);
	}

	@Override
	public Future<Entity> get(final Transaction txn, final Key key)
	{
		return exec.submit(new Callable<Entity>()
		{
			@Override
			public Entity call() throws Exception
			{
				try
				{
					Entity toReturn = get(txn, Collections.singleton(key)).get().values().iterator().next();
					if (toReturn == null)
					{
						throw new EntityNotFoundException(key);
					}
					return toReturn;
				}
				catch (NoSuchElementException npe)
				{
					throw new EntityNotFoundException(key);
				}
				catch (NullPointerException npe)
				{
					throw new EntityNotFoundException(key);
				}
			}
		});
	}

	@Override
	public Future<Map<Key, Entity>> get(final Transaction txn, final Iterable<Key> keys)
	{
		return exec.submit(new Callable<Map<Key, Entity>>()
		{
			@Override
			public Map<Key, Entity> call() throws Exception
			{
				HBaseTransaction thisTxn = checkTransaction(txn);
				//TODO: Chunking.
				//TODO: Implement multiple gets in HBaseTransaction?? This is far too synchronous.
				//TODO: Implement an asychronous getting service that reads in several threads?
				//TODO: Do this without HAcid if we're not in a transaction?
				Map<Key, Entity> toReturn = new HashMap<Key, Entity>();
				for (Key key : keys)
				{
					HAcidGet get = new HAcidGet(main(), key.toKeyString().getBytes());
					get.addFamily(FAMILY);
					Result res = null; //TODO: thisTxn.get(get);
					Map<byte[], byte[]> map = res.getFamilyMap(FAMILY);
					Entity entity = HBaseTranslator.toEntity(map, res.getRow());
					toReturn.put(entity.getKey(), entity);
				}
				if (thisTxn.isAutocommit())
				{
					thisTxn.commit();
				}
				return toReturn;
			}
		});
	}

	@Override
	public Collection<Transaction> getActiveTransactions()
	{
		return new ArrayList<Transaction>(transactions);
	}

	@Override
	public Transaction getCurrentTransaction()
	{
		return currentTxn;
	}

	@Override
	public Transaction getCurrentTransaction(Transaction txn)
	{
		if (currentTxn == null || !currentTxn.isActive())
		{
			return txn;
		}
		return currentTxn;
	}

	@Override
	public PreparedQuery prepare(Transaction txn, Query query)
	{
		return new HBaseScanPreparedQuery(txn, query, main(), index(), exec);
	}

	@Override
	public PreparedQuery prepare(Query query)
	{
		return prepare(null, query);
	}

	@Override
	public Future<Key> put(final Transaction txn, final Entity entity)
	{
		return exec.submit(new Callable<Key>()
		{
			@Override
			public Key call() throws Exception
			{
				return put(txn, Collections.singleton(entity)).get().iterator().next();
			}
		});
	}

	@Override
	public Future<List<Key>> put(Iterable<Entity> entities)
	{
		return put(null, entities);
	}

	@Override
	public Future<Key> put(Entity entity)
	{
		return put(null, entity);
	}

	@Override
	public Future<List<Key>> put(final Transaction txn, final Iterable<Entity> entities)
	{
		return exec.submit(new Callable<List<Key>>()
		{
			@Override
			public List<Key> call() throws Exception
			{
				HBaseTransaction thisTxn = checkTransaction(txn);
				//TODO: Chunking.
				//TODO: Implement multiple puts in HBaseTransaction?? This is far too synchronous.
				//TODO: Implement an asychronous putting service that writes in several threads?
				List<Key> toReturn = new ArrayList<Key>();
				for (Entity e : entities)
				{
					Key k = e.getKey();
					if (k == null || !k.isComplete())
					{
						String kind;
						Key parent = null;
						if (k == null)
						{
							kind = e.getKind();
						}
						else
						{
							kind = k.getKind();
							parent = k.getParent();
						}
						KeyRange range = allocateIds(parent, kind, 1l).get();
						k = range.iterator().next();
						e.setKey(k);
					}
					HAcidPut put = new HAcidPut(main(), k.toKeyString().getBytes());
					for (Map.Entry<String, Object> entry : e.getProperties().entrySet())
					{
						put.add(FAMILY, entry.getKey().getBytes(), HBaseTranslator.toBytes(entry.getValue()));
					}
					//TODO: thisTxn.put(put);
					toReturn.add(k);
				}
				indexer.index(entities);
				if (thisTxn.isAutocommit())
				{
					thisTxn.commit();
				}
				return toReturn;
			}
		});
	}

	@Override
	public Future<DatastoreAttributes> getDatastoreAttributes()
	{
		//This doesn't need to be asynchronous at all for us.
		return new FutureTask<DatastoreAttributes>(new Callable<DatastoreAttributes>()
		{
			@Override
			public DatastoreAttributes call()
			{
				return new DatastoreAttributes();
			}
		});
	}

	@Override
	public Future<Map<Index, IndexState>> getIndexes()
	{
		//TODO Implement getIndexes
		throw new UnsupportedOperationException("Not supported yet.");
	}

	@Override
	public Future<Transaction> beginTransaction()
	{
		return beginTransaction(null);
	}

	@Override
	public KeyRangeState allocateIdRange(KeyRange range)
	{
		//TODO: prevent auto allocation of keys in the range given.
		throw new UnsupportedOperationException("Not supported yet.");
	}
}
